package com.fastmms.analysis;

import com.fastmms.config.FastMmsConfig;
import com.fastmms.utils.ByteUtil;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

public class AnalysisFileUtil {

    private static ThreadLocal<AnalysisByte> FILE_THREAD_LOCAL = new ThreadLocal<>();

    public static void handleFile(String msgPath, AnalysisBlockResult allResult ,long startNo , int limit , Object param, AnalysisMessage analysisMessage) {
        AnalysisBlockResult indexResult = findBlock(allResult,startNo,limit);
        handleFile(msgPath,indexResult,param,analysisMessage);
    }

    public static void handleFile(String msgPath,AnalysisBlockResult indexResult,  Object param, AnalysisMessage analysisMessage) {
        File msgFile = new File(msgPath);
        if(!msgFile.exists() || msgFile.isDirectory()){
            return;
        }

        List<AnalysisBlock> analysisBlockList = null;
        AnalysisBlockResult analysisBlockResult;
        try {
            if( indexResult == null ){
                analysisBlockResult = analysisIndex( getIndexPath(msgPath) );
            }else {
                analysisBlockResult = indexResult;
            }
            analysisBlockList = analysisBlockResult.getAnalysisBlockList();
            analysisMessage.ready(  analysisBlockList , param);
        } catch (Exception e) {
            analysisMessage.error(null, -1, param, e);
            analysisMessage.finish(0, param);
            return;
        }

        if( analysisBlockList == null || analysisBlockList.size() == 0 ){
            analysisMessage.finish(0, param);
            return;
        }

        long readCount = 0;
        try {

            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(FastMmsConfig.ANALYSIS_MAXIMUM_POOL_SIZE , FastMmsConfig.ANALYSIS_MAXIMUM_POOL_SIZE
                    , 60L, TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>()
                    , Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

            List<AnalysisByte> analysisByteList = new ArrayList<>(FastMmsConfig.ANALYSIS_MAXIMUM_POOL_SIZE);
            CountDownLatch count = new CountDownLatch(analysisBlockList.size());
            int maxBlockSize = analysisBlockResult.getMaxBlockSize();
            long submit = 0;
            List<AnalysisData> analysisDataList = new LinkedList<>();
            ReentrantLock sortLock =  new ReentrantLock();

            long startNo = analysisBlockResult.getStartNo();
            long endNo = analysisBlockResult.getEndNo();

            for (AnalysisBlock block : analysisBlockList) {
                if(FastMmsConfig.SEQUENTIAL_ANALYSIS_MODEL && submit != 0 && submit % FastMmsConfig.ANALYSIS_MAXIMUM_POOL_SIZE == 0){
                    // 等待完成
                    while(true){
                        if(threadPoolExecutor.getActiveCount() <= 0){
                            break;
                        }
                    }
                    if(analysisDataList.size() > 0){
                        sortLock.lock();
                        Collections.sort(analysisDataList);
                        try {
                            for(AnalysisData analysisData : analysisDataList){
                                analysisMessage.apply(analysisData.getBlockNo() , analysisData.getMsgNo() , analysisData.getMsgLingNo()
                                        , analysisData.getKeyData()  ,analysisData.getValueData() , param);
                            }
                        }catch (Exception e){
                            e.printStackTrace();
                        }finally {
                            analysisDataList.clear();
                            sortLock.unlock();
                        }
                    }
                }
                submit++;
                threadPoolExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            AnalysisByte analysisByte = FILE_THREAD_LOCAL.get();
                            if(analysisByte == null){
                                RandomAccessFile msgFile = new RandomAccessFile(msgPath, "r");
                                analysisByte = new AnalysisByte(msgFile,maxBlockSize);
                                FILE_THREAD_LOCAL.set(analysisByte);
                                analysisByteList.add(analysisByte);
                            }
                            analysisByte.initData(block);
                            analysisBlock( analysisByte, analysisMessage,startNo,endNo, param,  sortLock , analysisDataList);
                        }catch (Exception e){
                            analysisMessage.error(null, -1, param, e);
                        }finally {
                            count.countDown();
                        }
                    }
                });
            }
            threadPoolExecutor.shutdown();
            try {
                // 等待全部执行完成
                count.await();
                // 让我们看看是否还有解析数据没有输出
                if(FastMmsConfig.SEQUENTIAL_ANALYSIS_MODEL && analysisDataList.size() > 0){
                    try {
                        for(AnalysisData analysisData : analysisDataList){
                            analysisMessage.apply(analysisData.getBlockNo() , analysisData.getMsgNo() , analysisData.getMsgLingNo()
                                    , analysisData.getKeyData()  ,analysisData.getValueData() , param);
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }finally {
                        analysisDataList.clear();
                    }
                }
                // 执行统计
                for (AnalysisBlock block : analysisBlockList) {
                    readCount += block.getMsgSize();
                }
                // 关闭全部文件资源
                for (AnalysisByte analysisByte : analysisByteList){
                    analysisByte.close();
                }
            } catch (Exception e) {
                analysisMessage.error(null, -1, param, e);
            }
        } finally {
            analysisMessage.finish(readCount, param);
        }

    }

    public static String getIndexPath(String msgPath){
        return msgPath.substring(0 , msgPath.lastIndexOf(".")) + FastMmsConfig.INDEX_SUFFIX;
    }

    /*public static AnalysisBlockResult findBlock(AnalysisBlockResult indexResult , long startNo , int limit ){
        AnalysisBlockResult findData = new AnalysisBlockResult();
        List<AnalysisBlock> analysisBlockList = new LinkedList<>();
        findData.setAnalysisBlockList(analysisBlockList);
        // 如果消息号小于0 则无法找到
        if(startNo <= 0 || limit <= 0){
            return findData;
        }
        // 如果索引数据没有，则无法找到
        List<AnalysisBlock> resultBlockList = indexResult.getAnalysisBlockList();
        if( resultBlockList == null || resultBlockList.size() == 0){
            return findData;
        }
        AnalysisBlock startBlock = resultBlockList.get(0);
        // 如果第一个消息块的消息号 大于需要查找的范围，则无法找到
        if( startBlock.getMsgStartNo() > startNo + limit ){
            return findData;
        }
        // 如果最后一块的消息号小于查找序号，则无法找到
        AnalysisBlock endBlock = resultBlockList.get(resultBlockList.size() - 1 );
        if(endBlock.getMsgStartNo() < startNo){
            return findData;
        }

        int startBlockNo = -1;
        int endBlockNo = -1;
        int lastIndex = 0;
        for (int i = 0; i < resultBlockList.size(); i++) {
            AnalysisBlock findBlock = resultBlockList.get(i);
            if( startBlockNo < 0 && startNo < findBlock.getMsgStartNo() ) {
                startBlockNo = lastIndex;
            }
            if( startBlockNo > -1 && startNo + limit < findBlock.getMsgStartNo() ){
                endBlockNo = lastIndex;
                break;
            }
            lastIndex = i;
        }
        if(startBlockNo > -1){
            long maxMsgSize = -1;
            endBlockNo = endBlockNo > -1 ?  endBlockNo : resultBlockList.size();
            for (int i = startBlockNo; i <= endBlockNo; i++) {
                AnalysisBlock findBlock = resultBlockList.get(i);
                maxMsgSize = Math.max(maxMsgSize,findBlock.getEndPoint() - findBlock.getStartPoint() );
                analysisBlockList.add(findBlock);
            }
            findData.setMaxBlockSize((int) maxMsgSize);
            findData.setStartNo(startNo);
            findData.setEndNo( startNo + limit - 1);
        }
        return findData;
    }*/

    public static AnalysisBlockResult findBlock(AnalysisBlockResult indexResult , long startNo , int limit ){
        AnalysisBlockResult findData = new AnalysisBlockResult();

        // 如果消息号小于0 则无法找到
        if(startNo <= 0 || limit == 0){
            return findData;
        }
        int tempStartNo = startNo > Integer.MAX_VALUE ? 0 : (int)startNo;
        limit = limit < 0 ?  Integer.MAX_VALUE - tempStartNo - 8 : limit;
        // 如果索引数据没有，则无法找到
        List<AnalysisBlock> resultBlockList = indexResult.getAnalysisBlockList();
        if( resultBlockList == null || resultBlockList.size() == 0){
            return findData;
        }
        AnalysisBlock startBlock = resultBlockList.get(0);
        // 如果第一个消息块的消息号 大于需要查找的范围，则无法找到
        if( startBlock.getMsgStartNo() > startNo + limit ){
            return findData;
        }
        // 如果最后一块的消息号小于查找序号，则无法找到
        AnalysisBlock endBlock = resultBlockList.get(resultBlockList.size() - 1 );
        if(endBlock.getMsgEndNo() < startNo){
            return findData;
        }
        AnalysisBlockResult.EntryNode findStartNode = null;
        AnalysisBlockResult.EntryNode findEndNode = null;
        long endNo = startNo + limit;
        List<AnalysisBlockResult.EntryNode> nodeList = indexResult.getNodeList();//查找一级索引
        for (AnalysisBlockResult.EntryNode node : nodeList){
            if( findStartNode == null && startNo >= node.getStartMsgNo() && startNo <= node.getEndMsgNo()){
                findStartNode = node;
            }
            if( findStartNode != null && endNo >= node.getStartMsgNo() && endNo <= node.getEndMsgNo() ){
                findEndNode = node;
                break;
            }
        }

        if(findStartNode == null){
            return findData;
        }
        // 获取一级范围
        int startIndex = findStartNode.getStartIndex();
        int endIndex = resultBlockList.size() - 1;
        if( findEndNode != null ){
            endIndex = findEndNode.getEndIndex();
        }

        List<AnalysisBlock> findList = resultBlockList.stream()
                .skip( startIndex )
                .limit( endIndex - startIndex + 1 )
                .collect(Collectors.toList());
        // 获取精确在那一块
        startIndex = findIndexSearch( findList , startNo );
        if(startIndex < 0){
            return findData;
        }
        findList = findList.stream().skip(startIndex).collect(Collectors.toList());
        endIndex = findIndexSearch( findList , endNo );
        if(endIndex > -1){
            findList = findList.stream().limit(endIndex+1).collect(Collectors.toList());
        }

        long maxMsgSize = -1;
        for(AnalysisBlock findBlock : findList){
            maxMsgSize = Math.max(maxMsgSize,findBlock.getEndPoint() - findBlock.getStartPoint() );
        }

        findData.setAnalysisBlockList(findList);
        findData.setMaxBlockSize((int) maxMsgSize);
        findData.setStartNo(startNo);
        findData.setEndNo( startNo + limit - 1);
        return findData;
    }

    public static int findIndexSearch(List<AnalysisBlock> resultBlockList , long findMsgNo ){
        if(resultBlockList == null || resultBlockList.size() == 0 || findMsgNo <= 0){
            return -1;
        }
        int left = 0;
        int right = resultBlockList.size() - 1;
        long startMsgNo;
        long endMsgNo;
        while( left + 1 < right ){
            int mid = left + (right - left) / 2 ;
            AnalysisBlock block = resultBlockList.get(mid);
            startMsgNo =  block.getMsgStartNo();
            endMsgNo =  block.getMsgEndNo();
            if(  findMsgNo > startMsgNo && findMsgNo < endMsgNo ){
                return mid;
            }else if( findMsgNo < startMsgNo){
                right = mid;
            }else {
                left = mid;
            }
        }
        AnalysisBlock result = resultBlockList.get(left);
        if( findMsgNo >= result.getMsgStartNo() && findMsgNo <= result.getMsgEndNo()){
            return left;
        }
        result = resultBlockList.get(right);
        if( findMsgNo >= result.getMsgStartNo() && findMsgNo <= result.getMsgEndNo() ){
            return right;
        }
        return -1;
    }

    /*public static void main(String[] args) {
        List<AnalysisBlock> list = new ArrayList<>(255);
        int point = 0;
        for (int i = 1; i <= 255; i++) {
            AnalysisBlock block = new AnalysisBlock();
            list.add(block);
            point += 1;
            block.setStartPoint(point);
            point += 10;
            block.setEndPoint(point);
        }
        AnalysisBlockResult result = new AnalysisBlockResult();
        result.setAnalysisBlockList(list);
        result.buildNode();
        List<AnalysisBlockResult.EntryNode> nodeList = result.getNodeList();
        for (AnalysisBlockResult.EntryNode node : nodeList){
            System.out.println(node);
        }
        //System.out.println(findIndexSearch(list,50));
    }*/


    public static AnalysisBlockResult analysisIndex(String idxPath) throws Exception {
        File idxFile = new File(idxPath);
        if (!idxFile.exists() || idxFile.isDirectory()) {
            throw new Exception("索引文件不存在：" + idxFile);
        }

        AnalysisBlockResult analysisBlockResult = new AnalysisBlockResult();
        int blockSize = (int) (idxFile.length() / 16); // 计算块数
        List<AnalysisBlock> blockList = new ArrayList<>(blockSize);
        analysisBlockResult.setAnalysisBlockList(blockList);

        FileInputStream idxIn = new FileInputStream(idxFile);
        byte[] idxByte = new byte[1024];
        int maxBlockSize = -1;
        int readSize;
        long startPoint = 0;
        long blockNo = 0;
        try {
            while( (readSize = idxIn.read(idxByte) ) > 0 ){
                int idxBytePoint = 0;
                while(idxBytePoint < readSize){
                    blockNo++;
                    long idx = ByteUtil.bytesToLongBig(idxByte,idxBytePoint);
                    idxBytePoint += 8;
                    long msgStartNo = ByteUtil.bytesToLongBig(idxByte,idxBytePoint);
                    idxBytePoint += 8;

                    AnalysisBlock block = new AnalysisBlock();
                    block.setStartPoint(startPoint);
                    block.setEndPoint(idx);
                    block.setMsgStartNo(msgStartNo);
                    block.setBlockNo(blockNo);
                    blockList.add(block);
                    startPoint = idx;
                    maxBlockSize = Math.max(maxBlockSize, (int)(block.getEndPoint() - block.getStartPoint()) );
                }
            }
        }finally {
            idxIn.close();
        }

        if(blockList.size() > 0){
            AnalysisBlock lastBlock = blockList.get( blockList.size() - 1 );
            lastBlock.setMsgEndNo(Integer.MAX_VALUE);
            if(blockList.size() > 1){
                for (int i = blockList.size() - 2; i >= 0 ; i--) {
                    AnalysisBlock block = blockList.get(i);
                    block.setMsgEndNo( lastBlock.getMsgStartNo() - 1 );
                    lastBlock = block;
                }
            }
        }

        analysisBlockResult.setMaxBlockSize(maxBlockSize);
        analysisBlockResult.buildNode();
        return analysisBlockResult;
    }

    private static void analysisBlock( AnalysisByte analysisByte , AnalysisMessage analysisMessage ,long startNo , long endNo
            , Object param , ReentrantLock sortLock,List<AnalysisData> analysisDataList) {
        long countSize = 0;
        AnalysisBlock block = analysisByte.getBlock();
        long blockNo = block.getBlockNo();
        int keySize;
        int valueSize;
        // 消息块开始行号
        long startMsgNo = block.getMsgStartNo();
        // 行标
        long msgLingNo = startMsgNo - 1;
        // 块内行号
        long msgNo = 0;

        try {
            while (analysisByte.hasNext()){
                msgLingNo++;
                msgNo++;
                if( startNo > 0 &&  startNo > msgLingNo ){
                    moveMsg( analysisByte);
                    continue;
                }
                if( endNo > 0 &&  msgLingNo > endNo ){
                    moveMsg( analysisByte);
                    continue;
                }

                keySize = analysisByte.readInt();
                byte[] keyData = new byte[keySize];
                analysisByte.read(keyData);

                valueSize = analysisByte.readInt();
                byte[] valueData = new byte[valueSize];
                analysisByte.read(valueData);

                if(FastMmsConfig.SEQUENTIAL_ANALYSIS_MODEL){
                    AnalysisData data = new AnalysisData();
                    data.setBlockNo(blockNo);
                    data.setMsgNo(msgNo);
                    data.setKeyData(keyData);
                    data.setValueData(valueData);
                    data.setMsgLingNo(msgLingNo);
                    sortLock.lock();
                    analysisDataList.add(data);
                    sortLock.unlock();
                }else {
                    analysisMessage.apply(blockNo, msgNo, msgLingNo ,keyData, valueData, param);
                }
                countSize++;
            }
        } catch (Exception e) {
            if (analysisByte.getMsgFile() != null) {
                try {
                    analysisMessage.error(block, analysisByte.getMsgFile().getFilePointer(), param, e);
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            } else {
                analysisMessage.error(block, 0, param, e);
            }
        } finally {
            block.setMsgSize(countSize);
            analysisByte.clearAnalysisData();
        }
    }

    private static void moveMsg(AnalysisByte analysisByte){
        int keySize = analysisByte.readInt();
        analysisByte.movePoint(keySize);
        int valueSize = analysisByte.readInt();
        analysisByte.movePoint(valueSize);
    }


}
